package net.wicp.tams.common.binlog.self;

import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.SocketUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.bean.CountNum;
import net.wicp.tams.common.binlog.self.bean.Host;
import net.wicp.tams.common.binlog.self.bean.PacketBean;
import net.wicp.tams.common.binlog.self.bean.Pos;
import net.wicp.tams.common.binlog.self.bean.RsCol;
import net.wicp.tams.common.binlog.self.constant.Capability;
import net.wicp.tams.common.binlog.self.constant.Checksum;
import net.wicp.tams.common.binlog.self.constant.Command;
import net.wicp.tams.common.binlog.self.reader.BinglogReadDump;
import net.wicp.tams.common.binlog.self.replication.down.AbsResultPacket;
import net.wicp.tams.common.binlog.self.replication.down.EOFPacket;
import net.wicp.tams.common.binlog.self.replication.down.ErrorPacket;
import net.wicp.tams.common.binlog.self.replication.down.GreetingPacket;
import net.wicp.tams.common.binlog.self.replication.down.OKPacket;
import net.wicp.tams.common.binlog.self.replication.down.RsFieldPacket;
import net.wicp.tams.common.binlog.self.replication.down.RsHeaderPacket;
import net.wicp.tams.common.binlog.self.replication.down.RsRowPacket;
import net.wicp.tams.common.binlog.self.replication.up.SecurePasswordAuthenticationPacket;
import net.wicp.tams.common.binlog.self.replication.up.command.AbsCommand;
import net.wicp.tams.common.binlog.self.replication.up.command.ComBinlogDump;
import net.wicp.tams.common.binlog.self.replication.up.command.ComBinlogDumpGtid;
import net.wicp.tams.common.binlog.self.replication.up.command.ComInitDb;
import net.wicp.tams.common.binlog.self.replication.up.command.ComQuery;
import net.wicp.tams.common.binlog.self.replication.up.command.SingleCommand;
import net.wicp.tams.common.binlog.self.sender.ISender;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.io.OutputStreamBufferedWrite;
import net.wicp.tams.common.thread.threadlocal.PerthreadManager;

@Slf4j
public class Dump {
	private final String host;
	private final int port;
	private Socket socket;
	OutputStreamBufferedWrite out;
	InputStream in;
	private boolean isLogin = false;
	private BinglogReadDump binlogParser;// 只有做了 dumpLog才有值

	public Dump(String host, int port) {
		this.host = host;
		this.port = port;
	}

	public Dump(String host) {
		this(host, 3306);
	}

	public Result login(String user, String pwd, String defaultSchema) throws IOException {
		try {
			if (socket == null || socket.isClosed()) {
				socket = SocketUtil.create(host, port);
			}
			out = new OutputStreamBufferedWrite(socket.getOutputStream());
			in = socket.getInputStream();
		} catch (Exception e) {
			throw new IllegalAccessError(String.format("不能连接到主机:[%s]，端口：[%s]", host, port));
		}
		PacketBean greatBean = PacketBean.buildBean(in);
		if (greatBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			final ErrorPacket error = new ErrorPacket(greatBean);
			return Result.getError("欢迎包错误").setRetObjs(error);
		} else {
			Host hostpre = (Host) PerthreadManager.getInstance().createValue("zorro-host",Host.class).get(new Host());
			hostpre.setHostIp(host);
			hostpre.setPort(port);
			hostpre.setUser(user);
			hostpre.setPwd(pwd);
			hostpre.setDefaultDb(defaultSchema);
			PerthreadManager.getInstance().createValue("zorro-host",Host.class).set(hostpre);
			GreetingPacket greetingPacket = new GreetingPacket(greatBean);
			SecurePasswordAuthenticationPacket pa = new SecurePasswordAuthenticationPacket(greetingPacket);
			pa.setUser(user);
			pa.setPassword(pwd);
			pa.setDefaultSchema(defaultSchema);
			byte[] bytes = pa.writeBody();
			out.write(bytes);
			out.flush();
			PacketBean loginRetBean = PacketBean.buildBean(in);
			if (loginRetBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
				final ErrorPacket error = new ErrorPacket(loginRetBean);
				return Result.getError("登陆失败").setRetObjs(error);
			} else if (loginRetBean.getPacketMarker() == OKPacket.PACKET_MARKER) {
				isLogin = true;
				OKPacket oKPacket = new OKPacket(loginRetBean);
				return Result.getSuc().setRetObjs(oKPacket);
			} else {
				return Result.getError(String.format("不支持的MARKER：[%s]", loginRetBean.getPacketMarker()))
						.setRetObjs(loginRetBean);
			}
		}
	}

	public void dumpLog(String binlogFilename, long binlogPos, long ServerId, boolean curFile, String dbPattern,
			String tbPattern, CountNum countNum, ISender... senders) throws IOException {
		dumpLogCommon(null, binlogFilename, binlogPos, ServerId, curFile, dbPattern, tbPattern, countNum, senders);
	}

	public void dumpLogGtid(String gtidStr, String binlogFilename, long binlogPos, long ServerId, boolean curFile,
			String dbPattern, String tbPattern, CountNum countNum, ISender... senders) throws IOException {
		if (StringUtil.isNull(gtidStr)) {
			throw new IllegalArgumentException("gtidStr不能为空");
		}
		dumpLogCommon(gtidStr, binlogFilename, binlogPos, ServerId, curFile, dbPattern, tbPattern, countNum, senders);
	}

	public void dumpLogGtid(String gtidStr, long ServerId, boolean curFile, String dbPattern, String tbPattern,
			CountNum countNum, ISender... senders) throws IOException {
		dumpLogGtid(gtidStr, "", 4, ServerId, curFile, dbPattern, tbPattern, countNum, senders);
	}

	/***
	 * 下载文件
	 * 
	 * @param binlogFilename
	 *            日志文件名
	 * @param binlogPos
	 *            日志文件位置
	 * @param ServerId
	 *            服务器ID
	 * @param curFile
	 *            是否只读当前文件 true:是，false：否
	 * @param dbPattern
	 *            库名模式
	 * @param tbPattern
	 *            表名模式
	 * @throws IOException
	 */

	private void dumpLogCommon(String gtidStr, String binlogFilename, long binlogPos, long ServerId, boolean curFile,
			String dbPattern, String tbPattern, CountNum countNum, ISender... senders) throws IOException {
		if (binlogPos < 4) {
			throw new IllegalAccessError("开始位置最小为4");
		}
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		long masterServerId = getMasterServerId();
		sendSql("SET @master_binlog_checksum='@@global.binlog_checksum'");
		sendSql(String.format("SET @mariadb_slave_capability='%s'", Capability.gtid.getValue()));
		String uuid = getUUID();
		// 设置全局配置信息
		Checksum checksum = getBinlogChecksum();

		boolean gtidCan = isAvailable(gtidStr, uuid);// 可以使用gtid
		log.info("gtid：[{}]是否可用：[{}]", gtidStr, gtidCan);
		AbsCommand pack = null;
		if (gtidCan) {// 是gtid
			ComBinlogDumpGtid comBinlogDumpGtid = new ComBinlogDumpGtid();
			GtidSet gtidSet = new GtidSet(gtidStr);
			comBinlogDumpGtid.setGtidSet(gtidSet);
			comBinlogDumpGtid.setBinlogFilename("");
			comBinlogDumpGtid.setBinlogPos(4);
			comBinlogDumpGtid.setServerId(ServerId);
			pack = comBinlogDumpGtid;
		} else {
			ComBinlogDump comBinlogDump = new ComBinlogDump();
			comBinlogDump.setBinlogFilename(binlogFilename);
			comBinlogDump.setBinlogPos(binlogPos);
			comBinlogDump.setFlags(0);
			comBinlogDump.setServerId(ServerId);
			pack = comBinlogDump;
		}

		byte[] input = pack.writeBody();
		out.write(input);
		out.flush();
		PacketBean binlogPacketBean = PacketBean.buildBean(in);// .buildEventBean(in,
																// binlogFilename);
		if (binlogPacketBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {// 第一个是ROTATE_EVENT事件
			final ErrorPacket binPacket = new ErrorPacket(binlogPacketBean);
			log.error("文件[{}],位置：[{}] 下载binlog时出错：[{}]", binlogFilename, binlogPos, binPacket.getBody().toJSONString());
		} else {
			log.info("下载binlog成功，文件名：[{}],位置：[{}],服务ID:[{}],内容：[{}]", binlogFilename, binlogPos, ServerId,
					binlogPacketBean.getBody());
			// 解析binlog日志
			binlogParser = new BinglogReadDump(uuid, checksum, in, dbPattern, tbPattern, senders);
			binlogParser.setFileName(binlogFilename);
			binlogParser.setCurFile(curFile);
			binlogParser.setMasterServerId(masterServerId);
			binlogParser.setGtids(gtidStr);
			if (countNum != null) {
				binlogParser.initCountNum(countNum);
			}
			try {
				binlogParser.read(binlogPos);
			} catch (Throwable e) {
				log.error("读binlog日志出现问题", e);
			}
		}
		destory();
	}

	public long getMasterServerId() throws IOException {
		String retstr = getVar("server_id", false);
		return Long.valueOf(retstr);
	}

	public YesOrNo getGtidSupport() throws IOException {
		String retstr = getVar("gtid_mode", true);
		return "ON".equals(retstr) ? YesOrNo.yes : YesOrNo.no;
	}

	public String getUUID() throws IOException {
		return getVar("server_uuid", false);
	}

	public String getGtidPurged() throws IOException {
		return getVar("GTID_PURGED", true);
	}

	public String getGtidExecuted() throws IOException {
		return getVar("GTID_Executed", true);
	}

	public String getVar(String var, boolean isGlobal) throws IOException {
		List<List<RsCol>> rslist = sqlQuery(String.format("select @@%s%s", isGlobal ? "GLOBAL." : "", var));
		RsCol rscol = rslist.get(0).get(0);
		return rscol.getValStr().replace("\n", "");
	}

	/***
	 * 得到当前的binlog_checksum
	 * 
	 * @return
	 * @throws IOException
	 */
	public Checksum getBinlogChecksum() throws IOException {
		List<List<RsCol>> rslist = sqlQuery("select @@binlog_checksum");
		RsCol rscol = rslist.get(0).get(0);
		String checksumStr = rscol.getValStr().replace("\n", "");
		return Checksum.get(checksumStr);
	}

	/***
	 * gitd能否可用
	 * 
	 * @param gtidStr
	 *            要验证的gtid
	 * @param uuid
	 *            当前uuid
	 * @return
	 * @throws IOException
	 */
	public boolean isAvailable(String gtidStr, String uuid) throws IOException {
		if (StringUtil.isNull(gtidStr)) {// 都没传gtid，说明不走gtid
			return false;
		}
		long maxGtidDel = maxGtid(getGtidPurged(), uuid);
		long maxGtidCur = maxGtid(gtidStr.replace("\n", ""), uuid);
		if (maxGtidDel == 0 || (maxGtidDel > 0 && maxGtidCur > 0 && maxGtidCur > maxGtidDel)) {
			return true;
		} else {
			return false;
		}
	}

	private long maxGtid(String gtidStr, String uuid) {
		if (StringUtil.isNull(gtidStr)) {// 本机在@@GLOBAL.GTID_PURGED为“”的情况
			return 0;
		}
		String[] gtidAry = gtidStr.split(",");
		long delmax = 0;
		for (String eleGtid : gtidAry) {
			if (eleGtid.startsWith(uuid)) {
				int index1 = eleGtid.lastIndexOf(":");
				String[] nums = eleGtid.substring(index1 + 1).split("-");
				String delNumStr = nums.length > 1 ? nums[1] : nums[0];
				delmax = Long.parseLong(delNumStr);
				break;
			}
		}
		return delmax;
	}

	public PacketBean sendSql(String sql) throws IOException {
		ComQuery common = new ComQuery();
		common.setSql(sql);
		byte[] input = common.writeBody();
		out.write(input);
		out.flush();
		PacketBean packet = PacketBean.buildBean(in);
		return packet;
	}

	/***
	 * 得到当前的位点
	 * 
	 * @return
	 */
	public Pos getDumpPos() {
		return binlogParser == null ? null : binlogParser.curPos();
	}

	public CountNum getCountNum() {
		return binlogParser == null ? null : binlogParser.getCountNum();
	}

	public BinglogReadDump getBinlogParser() {
		return this.binlogParser;
	}

	public void dumpLog(String binlogFilename, long binlogPos, long ServerId, String dbPattern, String tbPattern,
			CountNum countNum, ISender... senders) throws IOException {
		dumpLog(binlogFilename, binlogPos, ServerId, false, dbPattern, tbPattern, countNum, senders);
	}

	public List<List<RsCol>> sqlQuery(String sql) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		ComQuery common = new ComQuery();
		common.setSql(sql);
		byte[] input = common.writeBody();
		out.write(input);
		out.flush();
		PacketBean packet = PacketBean.buildBean(in);
		List<List<RsCol>> retlist = new ArrayList<>();
		if (packet.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			final ErrorPacket binPacket = new ErrorPacket(packet);
			log.error("查询时出错：[{}]", binPacket.getBody().toJSONString());
		} else {
			RsHeaderPacket rshead = new RsHeaderPacket(packet);
			System.out.println(rshead.getFieldCount());
			List<RsFieldPacket> fieldList = new ArrayList<>();
			while (true) {
				packet = PacketBean.buildBean(in);
				if (packet.getPacketMarker() == EOFPacket.PACKET_MARKER) {
					EOFPacket eof = new EOFPacket(packet);
					log.info("{}", eof);
					break;
				} else {
					RsFieldPacket field = new RsFieldPacket(packet);
					log.info("{}", field.getBody().toJSONString());
					fieldList.add(field);
				}
			}

			while (true) {
				packet = PacketBean.buildBean(in);
				if (packet.getPacketMarker() == EOFPacket.PACKET_MARKER) {
					EOFPacket eof = new EOFPacket(packet);
					log.info("{}", eof);
					break;
				} else {
					List<RsCol> rowvalue = new ArrayList<>();
					RsRowPacket row = new RsRowPacket(packet);
					log.info("{}", row.getBody().toJSONString());
					for (int i = 0; i < fieldList.size(); i++) {
						RsCol tempobj = RsCol.builder().columnType(fieldList.get(i).getFieldType())
								.valStr(row.getColumns().get(i)).build();
						rowvalue.add(tempobj);
					}
					retlist.add(rowvalue);
				}
			}
		}
		return retlist;
	}

	/***
	 * 发送单个命令
	 * 
	 * @param command
	 *            命令
	 * @return
	 * @throws IOException
	 */
	public AbsResultPacket sendSingleCommand(Command command) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		SingleCommand singleCommand = new SingleCommand(command);
		sendCommon(singleCommand);
		return retResult();
	}

	public AbsResultPacket setDefaultDb(String defaultDb) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		ComInitDb c = new ComInitDb();
		c.setDefaultSchema(defaultDb);
		sendCommon(c);
		return retResult();
	}

	private void sendCommon(AbsCommand command) throws IOException {
		byte[] input = command.writeBody();
		out.write(input);
		out.flush();
	}

	private AbsResultPacket retResult() throws IOException {
		PacketBean packet = PacketBean.buildBean(in);
		if (packet.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			return new ErrorPacket(packet);
		} else if (packet.getPacketMarker() == OKPacket.PACKET_MARKER) {
			return new OKPacket(packet);
		} else {
			return null;
		}
	}

	public void destory() throws IOException {
		isLogin = false;
		if (in != null) {
			in.close();
		}
		if (out != null) {
			out.close();
		}
		if (socket != null) {
			socket.close();
			socket = null;
		}
	}

}
